package com.gitee.xhs.window.lateness;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/7/27 12:42
 * @description OperatorLateDataFlatmap
 */
public class OperatorLateDataFlatMap implements FlatMapFunction<Transaction, Transaction> {

    private static final Long LAG_DATA_TIME_ONE_DAY = Time.days(1).toMilliseconds();
    private static final Long LAG_DATA_TIME_THIRTY_SECOND = Time.seconds(30).toMilliseconds();

    @Override
    public void flatMap(Transaction value, Collector<Transaction> out) throws Exception {
        out.collect(value);
        // 减少时间，形成迟到数据，1分钟
        value.setTimestamp(makeLagTime(value.getTimestamp(), LAG_DATA_TIME_THIRTY_SECOND));
        out.collect(value);
        // 减少时间，形成迟到数据, 1天
        value.setTimestamp(makeLagTime(value.getTimestamp(), LAG_DATA_TIME_ONE_DAY));
        out.collect(value);
    }

    private Long makeLagTime(Long elementTime, Long lagTime) {
        return elementTime - lagTime;
    }

}
